/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.remoting.rpc;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingInfo;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingUtils;

/**
 * 客户端元数据：
 * 1. brokerAddrTable：broker地址表；
 * 2. brokerVersionTable：broker版本表（没用）；
 * 3. topicEndPointsTable：主题端点表；
 * 4. topicRouteTable：主题路由表。
 */
public class ClientMetadata {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);

    private final ConcurrentMap<String/* Topic */, TopicRouteData> topicRouteTable = new ConcurrentHashMap<>();

    /**
     * 主题名称到消息队列集合的对应关系。
     * key为主题名称；
     * value为消息队列与broker名称的对应关系
     *
     * 即：一个主题横跨多个broker名称，每个broker名称上分配若干消息队列。
     * 一个broker名称对应一个主从集群，一个master，多个slave。
     *
     * MessageQueue表示主题的一个消息队列。
     * - 主题名称；
     * - broker名称；
     * - 消息队列ID。
     *
     */
    private final ConcurrentMap<String/* Topic */, ConcurrentMap<MessageQueue, String/*brokerName*/>> topicEndPointsTable = new ConcurrentHashMap<>();

    /**
     * ConcurrentHashMap对象
     * key是broker名称，value是HahsMap集合（key是brokerId，value是broker地址）
     */
    private final ConcurrentMap<String/* Broker Name */, HashMap<Long/* brokerId */, String/* address */>> brokerAddrTable =
        new ConcurrentHashMap<>();

    private final ConcurrentMap<String/* Broker Name */, HashMap<String/* address */, Integer>> brokerVersionTable =
        new ConcurrentHashMap<>();

    /**
     * 更新主题路由信息
     * @param topic 主题名称
     * @param topicRouteData 主题路由数据
     */
    public void freshTopicRoute(String topic, TopicRouteData topicRouteData) {
        // 如果主题名称或主题路由数据任何一个为null，则返回
        if (topic == null
            || topicRouteData == null) {
            return;
        }

        // 从主题路由表中获取指定主题的旧的路由数据
        TopicRouteData old = this.topicRouteTable.get(topic);
        // 如果新的主题路由信息和原来缓存的一致，则无需更新，直接返回
        if (!topicRouteData.topicRouteDataChanged(old)) {
            return ;
        }

        {
            // 遍历新的主题路由数据中的broker数据
            for (BrokerData bd : topicRouteData.getBrokerDatas()) {
                // 更新broker地址表
                this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());
            }
        }
        {
            //  遍历新的主题路由数据中的broker数据
            ConcurrentMap<MessageQueue, String> mqEndPoints = topicRouteData2EndpointsForStaticTopic(topic, topicRouteData);
            if (mqEndPoints != null
                    && !mqEndPoints.isEmpty()) {
                // 将主题的名称为key，mqEndPoints为value，放入主题路由表中
                topicEndPointsTable.put(topic, mqEndPoints);
            }
        }
    }

    /**
     * 从MessageQueue中获取broker的名称
     * - 首先从主题端点表中找出该mq对应的brokerName；
     * - 如果找不到，就直接返回mq中包含的brokerName。
     * @param mq 目标MessageQueue
     * @return 返回brokerName。
     */
    public String getBrokerNameFromMessageQueue(final MessageQueue mq) {
        // 如果主题端点表中包含该主题，并且mq与broker的映射不是空的
        if (topicEndPointsTable.get(mq.getTopic()) != null
                && !topicEndPointsTable.get(mq.getTopic()).isEmpty()) {
            // 通过主题名称获取mq与brokerName的映射，从中找出与指定mq对应的brokerName，返回
            return topicEndPointsTable.get(mq.getTopic()).get(mq);
        }
        // 否则直接返回mq中包含的brokerName值。
        return mq.getBrokerName();
    }

    /**
     * 更新集群信息
     * @param clusterInfo
     */
    public void refreshClusterInfo(ClusterInfo clusterInfo) {
        if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
            return;
        }
        // 遍历响应中的broker地址表
        for (Map.Entry<String, BrokerData> entry : clusterInfo.getBrokerAddrTable().entrySet()) {
            // 更新broker地址表
            brokerAddrTable.put(entry.getKey(), entry.getValue().getBrokerAddrs());
        }
    }

    /**
     * 根据brokerName获取master节点的地址
     * @param brokerName broker名称
     * @return 如果broker地址表中不包含该broker名称，则返回null；否则直接获取id为0的节点地址并返回。
     */
    public String findMasterBrokerAddr(String brokerName) {
        // 如果broker地址表中不包含该brokerName的条目，就返回null。
        if (!brokerAddrTable.containsKey(brokerName)) {
            return null;
        }
        // 否则根据brokerName从broker地址表中找出该brokerName对应的一组broker信息，然后找出id为0的broker的地址并返回。
        return brokerAddrTable.get(brokerName).get(MixAll.MASTER_ID);
    }

    public ConcurrentMap<String, HashMap<Long, String>> getBrokerAddrTable() {
        return brokerAddrTable;
    }

    /**
     * 将TopicRouteData转换为endpointsForStaticTopic
     * 返回值类型是：《ConcurrentMap<MessageQueue, String>，key是mq实例，value是brokerName》
     * @param topic
     * @param route 用于记录主题路由信息的类：
     *                  有序主题配置
     *                  主题的消息队列集合；
     *                  主题的broker集合；
     *                  主题在broker上的broker地址与主题队列的映射关系；
     *                  主题在每个所在的broker上的过滤器列表。
     * @return
     */
    public static ConcurrentMap<MessageQueue, String> topicRouteData2EndpointsForStaticTopic(final String topic, final TopicRouteData route) {
        // 如果主题路由数据中不包含主题根据broker的队列映射，或者该集合是空的
        // Map<String/*brokerName*/, TopicQueueMappingInfo> topicQueueMappingByBroker
        // TopicQueueMappingByBroker中的TopicQueueMappingInfo：主题到队列的映射信息；
        // 包含了主题名称（冗余），broker名称（主从关系），在当前broker上的队列总数，用于隔离旧的脏数据的纪元数字等。
        if (route.getTopicQueueMappingByBroker() == null
                || route.getTopicQueueMappingByBroker().isEmpty()) {
            // 返回空的map集合
            return new ConcurrentHashMap<>();
        }

        // 创建新的map集合
        ConcurrentMap<MessageQueue, String> mqEndPointsOfBroker = new ConcurrentHashMap<>();

        // 创建新的map集合
        Map<String, Map<String, TopicQueueMappingInfo>> mappingInfosByScope = new HashMap<>();

        // 遍历topicQueueMappingByBroker中的brokerName->TopicQueueMappingInfo映射条目
        for (Map.Entry<String, TopicQueueMappingInfo> entry : route.getTopicQueueMappingByBroker().entrySet()) {
            TopicQueueMappingInfo info = entry.getValue();
            String scope = info.getScope();
            if (scope != null) {
                if (!mappingInfosByScope.containsKey(scope)) {
                    mappingInfosByScope.put(scope, new HashMap<>());
                }
                mappingInfosByScope.get(scope).put(entry.getKey(), entry.getValue());
            }
        }

        for (Map.Entry<String, Map<String, TopicQueueMappingInfo>> mapEntry : mappingInfosByScope.entrySet()) {
            String scope = mapEntry.getKey();
            Map<String, TopicQueueMappingInfo> topicQueueMappingInfoMap =  mapEntry.getValue();
            ConcurrentMap<MessageQueue, TopicQueueMappingInfo> mqEndPoints = new ConcurrentHashMap<>();
            List<Map.Entry<String, TopicQueueMappingInfo>> mappingInfos = new ArrayList<>(topicQueueMappingInfoMap.entrySet());
            mappingInfos.sort((o1, o2) -> (int) (o2.getValue().getEpoch() - o1.getValue().getEpoch()));
            int maxTotalNums = 0;
            long maxTotalNumOfEpoch = -1;
            for (Map.Entry<String, TopicQueueMappingInfo> entry : mappingInfos) {
                TopicQueueMappingInfo info = entry.getValue();
                if (info.getEpoch() >= maxTotalNumOfEpoch && info.getTotalQueues() > maxTotalNums) {
                    maxTotalNums = info.getTotalQueues();
                }
                for (Map.Entry<Integer, Integer> idEntry : entry.getValue().getCurrIdMap().entrySet()) {
                    int globalId = idEntry.getKey();
                    MessageQueue mq = new MessageQueue(topic, TopicQueueMappingUtils.getMockBrokerName(info.getScope()), globalId);
                    TopicQueueMappingInfo oldInfo = mqEndPoints.get(mq);
                    if (oldInfo == null ||  oldInfo.getEpoch() <= info.getEpoch()) {
                        mqEndPoints.put(mq, info);
                    }
                }
            }


            //accomplish the static logic queues
            for (int i = 0; i < maxTotalNums; i++) {
                MessageQueue mq = new MessageQueue(topic, TopicQueueMappingUtils.getMockBrokerName(scope), i);
                if (!mqEndPoints.containsKey(mq)) {
                    mqEndPointsOfBroker.put(mq, MixAll.LOGICAL_QUEUE_MOCK_BROKER_NAME_NOT_EXIST);
                } else {
                    // key是MessageQueue实例，value是brokerName
                    mqEndPointsOfBroker.put(mq, mqEndPoints.get(mq).getBname());
                }
            }
        }
        return mqEndPointsOfBroker;
    }

}
